package com.runa.monitor.platinterface.mqtt;

import com.alibaba.fastjson.JSON;
import com.runa.persistence.util.LogUtil;
import com.runa.protocol.util.ByteUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.GZIPInputStream;

import static com.runa.monitor.platinterface.mqtt.MqttInitService.getTopicsList;

@Slf4j
public class BuildPublishConsumer implements MqttCallbackExtended {

    private MQTTExecutor mqttExecutor;

    private MQTTAsyncClient mqttAsyncClient;

    private AtomicInteger publishedCount = new AtomicInteger(0);

    public BuildPublishConsumer(MQTTExecutor mqttExecutor, MQTTAsyncClient mqttAsyncClient) {
        this.mqttExecutor = mqttExecutor;
        this.mqttAsyncClient = mqttAsyncClient;
    }

    //    @Resource
//    private CallBackHandler callBackHandler;

    /**
     * 断开重连
     *
     * @param cause
     */
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        log.error("连接断开，正在重连....", cause);
        publishedCount = new AtomicInteger(0);
        mqttAsyncClient.connectionLost();
    }


    @Override
    public void connectComplete(boolean b, String s) {
        log.info("重连成功");
        mqttAsyncClient.sub(getTopicsList());
    }

    /**
     * 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用
     *
     * @param token
     */
    public void deliveryComplete(IMqttDeliveryToken token) {
        publishedCount.incrementAndGet();
        mqttAsyncClient.publishComplete(token);
    }

    public int getPublishCompleteCount() {
        return publishedCount.get();
    }


    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        log.debug("{}: {}", topic, JSON.toJSONString(mqttMessage));
        messageReceived(topic, mqttMessage);
    }

    public String getId(String topic) {
        if (topic.startsWith("/")) {
            topic = topic.substring(1);
        }
        String[] split = topic.split("/");
        return split.length >= 2 ? split[1] : null;
    }

    public void messageReceived(String topic, MqttMessage mqttMessage) {
        LogUtil.getLogAndInfo(getId(topic), "MQTT", "{} {} {} {} {}", topic, mqttMessage.getId(), mqttMessage.getQos(), mqttMessage.isRetained(), getTransfer(mqttMessage.getPayload()));
        if (mqttMessage.isRetained()) {
            return;
        }
        if (topic.startsWith("/RU-LDTU-01")) {
            if (mqttMessage.getPayload().length > 7) {
                if ("func:up".equals(new String(mqttMessage.getPayload(), 0, 7))) {
                    String msg = new String(mqttMessage.getPayload());
                    mqttExecutor.sendLDTU(msg, null);
                    log.info("LDTU: {} 消息: {}", topic, msg);
                } else if (mqttMessage.getPayload()[0] == 0x7b &&
                        mqttMessage.getPayload()[mqttMessage.getPayload().length - 1] == 0x7b) {
                    mqttExecutor.sendLDTU(null, mqttMessage.getPayload());
                    log.info("LDTU: {} 消息: {}", topic, ByteUtil.byteToHexString(mqttMessage.getPayload()));
                }
            } else {
                log.info("LDTU: {} 消息: {}", topic, new String(mqttMessage.getPayload()));
            }
        } else if (topic.startsWith("/hotmeter-00")) {
            if (mqttMessage.getPayload().length > 7) {
                String msg = new String(mqttMessage.getPayload());
                mqttExecutor.sendHotmeter(msg);
                log.info("hotmeter: {} 消息: {}", topic, msg);
            } else {
                log.info("hotmeter: {} 消息: {}", topic, new String(mqttMessage.getPayload()));
            }
        } else {
            String json = "";
            try {
                json = getTransfer(mqttMessage.getPayload());
            } catch (Exception e) {
                log.error("Build Publish Consumer Accept: {}", json, e);
            }
            log.info("网关消息Build Publish Consumer Accept: {}", json);
            mqttExecutor.sendSms(topic, json);
        }
    }

    private String getTransfer(byte[] transfer) {
        try {
            if ("zip:".equals(new String(transfer, 0, 4))) {
                GZIPInputStream inputStream = new GZIPInputStream(new ByteArrayInputStream(transfer, 4, transfer.length - 4));
                ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                byte[] buff = new byte[256];
                int n;
                while ((n = inputStream.read(buff)) >= 0) {
                    outputStream.write(buff, 0, n);
                }
                return outputStream.toString();
            } else if (transfer[0] == (byte)0xfe) {
                return ByteUtil.byteToHexString(transfer);
            } else {
                return new String(transfer);
            }
        } catch (Exception e) {
            log.error("getTransfer: ", e);
            return null;
        }
    }

    public static void main(String[] args) throws IOException {
//        MQTTExecutor executor = new MQTTExecutor();
//        executor.sendSms("", "{\"Id\":\"14181006\",\"Func\":\"down_resp\",\"Seq\":\"1896234-{\\\"planTime\\\":1732095252349,\\\"taskType\\\":1}-HVA4840\",\"Time\":1732096593,\"Result\":{\"Data\":{\"Com\":true,\"Data\":{\"Id\":\"24808453\",\"Offer\":\"11-01~04-01\",\"Rl\":0,\"Kf\":110125,\"Interval\":0,\"TcMode\":0,\"Tactics\":1,\"TacticsValue\":0,\"DeglLimit\":0,\"DeghLimit\":0,\"VlckFlag\":1,\"VlckDst\":41,\"TlckFlag\":0,\"TlckDst\":25,\"TemplLimit\":5,\"TemphLimit\":45,\"TempCal\":0,\"TempDiff\":0,\"ChargeTip\":0,\"OwnTip\":0,\"ShowRl\":0,\"ShowKf\":0,\"SupTempCal\":0,\"RetTempCal\":0,\"OnAct\":0,\"DownAct\":0,\"SupTemp\":99.9,\"RetTemp\":35.5,\"WorkTime\":0,\"DateTime\":\"2024-11-20T17:56:40+08:00\",\"Staword\":\"0400\",\"SupportDeg\":1,\"RealDeg\":41,\"Version\":53,\"Dn\":3,\"SetTemp\":35.5,\"Rssi\":\"\",\"ReadTime\":\"2024-11-20T17:56:33+08:00\",\"Status\":\"\",\"HouseTempData\":{\"Id\":\"24808453\",\"CurTemp\":22.1,\"DateTime\":\"0001-01-01T00:00:00Z\",\"TpVersion\":0,\"TpBatteryVol\":3.6,\"TpCnt\":9,\"TpStateword\":\"48\",\"Rssi\":\"\",\"ReadTime\":\"2024-11-20T17:56:33+08:00\",\"Status\":\"\"}},\"Type\":\"housevalve\"},\"Type\":\"rddevice\"}}");
        String s = "{\"duplicate\":false,\"id\":0,\"messageId\":0,\"payload\":\"ZnVuYzp1cCxpZDoyNDMzMzgwNix0eXA6MSxuYmF0dGFjaDE6MCxyeHc6NixuYmRseToxLHJlcDo3NDAsY3ByZDowLHJyZW06MzAwLGNzcTowLGljYzogODk4NjA4MzIxMTIzRDAxMDU1MzANLGltZWk6ODY0MTQ1MDc0NDc1MjUwLGdwczoscnNycDotMTA3LHNucjo5LG9sZHM6M0YsbXZlcjowLGNpOjEzNDg2NTU3Nyxsb2M6MjIwNDUsNGd0cnk6MCw0Z21vZDogRUM4MDFFQ05DR1IwM0EwOE0wMg0sc2V0OjI1MCxzZXRzcmM6MCxjdXI6MjUwLjAsc2V0MDoxMDAsb3V0OjZ8MzAwLHJjbnQ6MCxtb2RlOjEsdGFjOjAsYWNsczoxLGFjbHQ6NDMyMDAsYmxvY2s6MCxibG9ja2RlYWw6MCxtb3RvczowLGNscmNudDowLG90aW1lOjAsZDFrOjUsY29uc3RjdXJ2ZTowLHQxOls5OS45XSx0MjpbOTkuOV0scDE6WzkuOTk5XSxwMjpbOS45OTldLHN0YTo4OTgwLHBhcjowMjAwLHBhcjI6MDAwMCx2ZXI6NzQzNSxydGM6MjAyNDEyMTExNTM0NDgsdWl0djE6MTAsdWl0djI6Mjg4MCxhaGVhZDoxMjAsaG90OjExMDEwNDAxLGJhdDozLjYwNSxsaWZlOjIzNTA3LHQxYzowLjAsdDJjOjAuMCxwMWM6MC4wLHAyYzowLjAscmVzZXQ6MixzZXR2Oi0xLjAsbG9nd3BvczoyNDE0LGVsb2d3cG9zOjAsd2t0bTo3MDg0MSx3a3RtMjo3NDAscmZsYWc6NDAscHNlbDowLHBkbHk6NCxtZHQ6MjQxMjExMTUzNDM5LG1vazoxLG1qcmw6MC4wMCxtcmw6MC4zMixtbGw6MTIuNjEsbXJnbDowLjAwLG1sc2w6MC4wMDAwLG10MToyMC44MSxtdDI6MjAuMzMsbXN0OjAwODAsbHBhZDo1NjcyLGxjbnQ6MCxvZHNlYzo3NCxyZGlmZjowLHJlcGVycmNudDowLGNlbnJlcHRpbWU6MTAsbGxpbWl0OjAsY2FsaWJlcnI6MCxtaW5rZDo1LGNvbXBpbGU6MjAyNC0wOS0yNFQwOS00MS0yMA0K\",\"qos\":0,\"retained\":false}";
        MqttMessage mqttMessage = JSON.parseObject(s, MqttMessage.class);
        BuildPublishConsumer buildPublishConsumer = new BuildPublishConsumer(null, null);
        String transfer = buildPublishConsumer.getTransfer(mqttMessage.getPayload());
        System.out.println(transfer);
    }

}
